package cn.kgc.listener;

import cn.kgc.entity.MessageInfo;
import cn.kgc.service.MessageInfoService;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;

/**
 * @author YC
 */
public abstract class AbstractMessageListener implements ChannelAwareMessageListener {

    @Resource
    private MessageInfoService messageInfoService;
    @Value("${mq.maxRetryCount}")
    private int maxRetryCount;

    /**
     * 消费消息的逻辑（业务处理逻辑）
     * 每个项目的处理逻辑和方法不同，设计为抽象方法，由项目继承此抽象类实现此抽象方法后自由编写消费逻辑
     *
     * @param message 消息
     * @param channel 通道
     * @throws Exception 异常
     */
    public abstract void receiveMessage(Message message, Channel channel) throws Exception;

    /**
     * 消息监听
     *
     * @param message 消息
     * @param channel 通道
     * @throws Exception 异常
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        // 获取消息相关属性
        String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        String exchange = message.getMessageProperties().getReceivedExchange();
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        String jsonStr = new String(message.getBody());
        String queue = message.getMessageProperties().getConsumerQueue();
        long delayTime = message.getMessageProperties().getHeaders().get("delayTime") == null
                ? 0L : (long) message.getMessageProperties().getHeaders().get("delayTime");
        // 判断消息是否第一次消费，是则持久化数据，否则更新消费次数
        MessageInfo messageInfo =
                messageInfoService.messageReceiveReady(messageId, exchange, routingKey, queue, jsonStr, delayTime);
        long tag = message.getMessageProperties().getDeliveryTag();
        // 判断是否超过消费次数最大次数，是则移入死信队列，否则正常消费信息
        if (messageInfo.getSequence() > maxRetryCount) {
            channel.basicNack(tag, false, false);
        }
        // 正常消费信息
        try {
            receiveMessage(message, channel);
            // 消费者确认处理消息
            channel.basicAck(tag, true);
            messageInfoService.messageDelete(messageId);
        } catch (Exception e) {
            // 如果产生异常（没有正确消费），将消息重新返回到消息队列等待重新消费
            Thread.sleep(1000L * messageInfo.getSequence());
            channel.basicNack(tag, false, true);
        }
    }

}
